package com.woniu.threaddemo;

import javax.validation.constraints.NotNull;
import java.util.Collection;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

public class ThreadPoolUtil {

    // 私有构造函数，防止实例化
    private ThreadPoolUtil() {
    }

    private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(
            4, // 核心线程数
            10, // 最大线程数
            60L, // 空闲线程等待时间
            TimeUnit.SECONDS, // 等待时间的单位
            new ArrayBlockingQueue<>(100), // 任务队列
            new ThreadFactory() {
                private int counter = 0;

                @Override
                public Thread newThread(@NotNull Runnable runnable) {
                    String prefix = "CustomPoolThread-";
                    return new Thread(runnable, prefix + counter++);
                }
            }, // 自定义线程工厂
            new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
    );

    // 1. 创建并行任务并执行等待结果返回
    public static <S, T> List<T> executeParallelTasksWithResult(Collection<S> data,
                                                                Function<S, T> task,
                                                                BiFunction<Throwable, S, T> exceptionHandler) {
        List<CompletableFuture<T>> futures = data.stream()
                .map(item -> CompletableFuture.supplyAsync(() -> task.apply(item), executor)
                        .exceptionally(ex -> exceptionHandler.apply(ex, item)))
                .collect(Collectors.toList());

        return futures.stream()
                .map(CompletableFuture::join)
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
    }

    // 2. 创建并行任务并执行无需等待结果返回
    public static <S> void executeParallelTasksWithoutResult(Collection<S> data,
                                                             Function<S, ?> task,
                                                             BiFunction<Throwable, S, ?> exceptionHandler) {
        data.forEach(item -> CompletableFuture.runAsync(() -> task.apply(item), executor)
                .exceptionally(ex -> {
                    exceptionHandler.apply(ex, item);
                    return null;
                }));
    }

    // 3. 创建单个任务并执行等待结果返回
    public static <T> T executeSingleTaskWithResult(Supplier<T> task, Function<Throwable, T> exceptionHandler) {
        return CompletableFuture.supplyAsync(task, executor)
                .exceptionally(exceptionHandler)
                .join();
    }

    // 4. 创建单个任务并执行无需等待结果返回
    public static void executeSingleTaskWithoutResult(Runnable task, Function<Throwable, Void> exceptionHandler) {
        CompletableFuture.runAsync(task, executor)
                .exceptionally(ex -> {
                    exceptionHandler.apply(ex);
                    return null;
                });
    }
}
